package com.jun.common.redis.stream;

import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author yuejun.li
 * @date 2024/8/13 21:54
 * redis stream消费者
 */
public abstract class RedisStreamConsumer implements StreamListener<String, MapRecord<String, String, String>> {

    private StreamMessageListenerContainer container;

    private RedisTemplate redisTemplate;

    public RedisStreamConsumer(StreamMessageListenerContainer streamMessageListenerContainer,RedisTemplate redisTemplate) {
        this.container = streamMessageListenerContainer;
        this.redisTemplate = redisTemplate;
    }

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        onStreamMessage(message);
    }

    /**
     * 配置消费组
     * @param consumer 消费者信息
     * @param streamOffset 队列的偏移量
     * @param isAutoAck 是否自动应答
     */
    public void receive(Consumer consumer,StreamOffset streamOffset,boolean isAutoAck){
        //消费组
        if(!isStreamGroupExists(redisTemplate,streamOffset.getKey().toString(),consumer.getGroup())){
            //左边是queueKey，右边是消费者组
            redisTemplate.opsForStream().createGroup(streamOffset.getKey().toString(),consumer.getGroup());
        }

        //是否自动作答
        if(isAutoAck){
            container.receiveAutoAck(consumer, streamOffset,this);
        }else{
            container.receive(consumer, streamOffset,this);
        }
    }

    protected abstract void onStreamMessage(MapRecord<String, String, String> message);

    /**
     * 判断该消费组是否存在
     * @param streamKey queue的key
     * @param groupName group的key
     * @return
     */
    public boolean isStreamGroupExists(RedisTemplate redisTemplate, String streamKey, String groupName) {
        //首先检查Stream Key是否存在，否则下面代码可能会因为尝试检查不存在的Stream Key而导致异常
        if (!redisTemplate.hasKey(streamKey)){
            return false;
        }
        //获取streamKey下的所有groups
        StreamInfo.XInfoGroups xInfoGroups = redisTemplate.opsForStream().groups(streamKey);
        AtomicBoolean exists= new AtomicBoolean(false);
        xInfoGroups.forEach(xInfoGroup -> {
            if (xInfoGroup.groupName().equals(groupName)){
                exists.set(true);
            }
        });
        return exists.get();
    }
}
